package prictice.trident.functions;

import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;
import prictice.base.DateFmt;

/**
 * @Package prictice.trident.functions
 * @Description:
 * @Author elwyn
 * @Date 2017/9/19 21:20
 * @Email elonyong@163.com
 */
public class MySplit extends BaseFunction {
    private static final long serialVersionUID = 82362714377935045L;
    private String patton = null;

    public MySplit(String patton) {
        this.patton = patton;
    }

    @Override
    public void execute(TridentTuple tridentTuple, TridentCollector tridentCollector) {
        String log = tridentTuple.getString(0);
        String[] logArr = log.split(patton);
        if (logArr.length == 3) {
            tridentCollector.emit(new Values(DateFmt.getCountDate(logArr[2], DateFmt.date_short), "cf", "pv_count", logArr[1]));
        }
    }
}
